package com.example.threadmodel.service;

import com.example.threadmodel.model.Action;
import com.example.threadmodel.model.ActionNoUidException;
import com.example.threadmodel.model.ActionQueue;
import com.example.threadmodel.model.NamedPoolThreadFactory;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Service
public class ThreadManagerService {
    private final static int THREAD_NUM = 32;
    public final static String ACTION_THREAD_NAME_PREFIX = "ACTION_THREAD-";

    @Getter
    private ExecutorService actionExecutors;

    private ScheduledExecutorService singleThreadScheduler;

    //<玩家id, 队列映射>
    private final ConcurrentHashMap<String, ActionQueue> uid2ActionQueueMap = new ConcurrentHashMap<>();

    // 暴露给监控，看看所有线程还没执行完的总数量
    @Getter
    private final AtomicLong userQueueSize = new AtomicLong();

    public void init() {
        singleThreadScheduler = Executors.newSingleThreadScheduledExecutor(new NamedPoolThreadFactory("ACTION-SINGLE"));

        actionExecutors = new ThreadPoolExecutor(THREAD_NUM, THREAD_NUM, 0, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(), new NamedPoolThreadFactory(ACTION_THREAD_NAME_PREFIX));

        // 15min检测一次(1.移除空队列 2.上报监控当前没执行完的数量)
        singleThreadScheduler.scheduleAtFixedRate(() -> {
            try {
                long removeTime = System.currentTimeMillis() - 15 * 60 * 1000;
                long sum = 0;

                for (Map.Entry<String, ActionQueue> entry : uid2ActionQueueMap.entrySet()) {
                    ActionQueue queue = entry.getValue();
                    sum += queue.getQueueSize();
                    if (!queue.isRunning()
                            && queue.getStopTimestamp() < removeTime
                            && queue.isEmpty()) { // 队列中没有数据了
                        removeQueue(entry.getKey(), removeTime); // 防止过多的空队列(比如：玩家下线了，任务也执行完了)
                    }
                }
                userQueueSize.set(sum); // 队列中总任务数, 这个放到监控里!!!

                // log.info("userQueueSize={}", userQueueSize);
            } catch (Throwable e) {
                log.error("error=", e);
            }

        }, 60, 60, TimeUnit.SECONDS);
    }

    /**
     * 核心方法, 从netty中取出任务，入队列
     *
     * @param action
     * @param <V>
     */
    public <V> void enqueueAction(Action<V> action) {
        // 绑定哪个队列
        String actionUid = action.getActionUid();

        //必须指定队列id
        if (actionUid == null) {
            throw new ActionNoUidException();
        }

        // 入队列
        uid2ActionQueueMap.compute(actionUid, (k, actionQueue) -> {
            // 队列
            if (actionQueue == null) {
                actionQueue = new ActionQueue(actionUid);
            }

            // action也知道当前是在哪个队列
            action.setActionQueue(actionQueue);

            // 提交任务，并开始执行
            actionQueue.offer(action);

            return actionQueue;
        });
    }

    private void removeQueue(String uid, long removeTime) {
        uid2ActionQueueMap.compute(uid, (k, queue) -> {
            if (queue == null) {
                return null;
            }
            // queue.run=true，仅在queue.offer()->startUpQueue()时发生；而queue.offer又仅在map.compute中调用
            // 因此!queue.isRun()的判断是安全的；而queue.isEmpty()为冗余校验
            if (!queue.isRunning()
                    && queue.getStopTimestamp() < removeTime
                    && queue.isEmpty()) {
                return null;
            }
            return queue;
        });
    }
}
